package com.shujia

import org.junit.{Before, Test}

import java.text.SimpleDateFormat
import java.util.Date
import scala.io.{BufferedSource, Source}

class Demo24Ceshi {
  case class LiveTypes(liveId: String, liveType: String)

  case class LiveEvents(userId: String, liveId: String, startTime: String, endTime: String)

  case class UserInfo(userId: String, userName: String)

  var liveTypes: List[LiveTypes] = _
  var liveEvents: List[LiveEvents] = _
  var userInfo: List[UserInfo] = _

  var userInfoMap: Map[String, String] = _
  var liveTypesMap: Map[String, String] = _

  def readFile(path: String): List[String] = {
    val bs: BufferedSource = Source.fromFile(path)
    val list: List[String] = bs.getLines().toList

    bs.close()

    list
  }

  @Before
  def init(): Unit = {
    liveTypes = readFile("data/ceshi/live_types.txt")
      .map(line => {
        val splits: Array[String] = line.split("\t")
        LiveTypes(splits(0), splits(1))
      })
    liveEvents = readFile("data/ceshi/live_events.txt")
      .map(line => {
        val splits: Array[String] = line.split("\t")
        LiveEvents(splits(0), splits(1), splits(2), splits(3))
      })
    userInfo = readFile("data/ceshi/user_info.txt")
      .map(line => {
        val splits: Array[String] = line.split("\t")
        UserInfo(splits(0), splits(1))
      })

    userInfoMap = userInfo.map(u => (u.userId, u.userName)).toMap
    liveTypesMap = liveTypes.map(t => (t.liveId, t.liveType)).toMap

  }

  @Test
  // 1、统计每位用户观看不同类型直播的次数（6分）
  // 输出：[用户id，用户名，直播间类型，次数]
  def question01(): Unit = {



    liveEvents
      .groupBy(e => e.userId + "|" + e.liveId)
      .map(kv => {
        val splits: Array[String] = kv._1.split("\\|")
        val userId: String = splits(0)
        val liveId: String = splits(1)

        val liveName: String = liveTypesMap.getOrElse(liveId, "")
        val userName: String = userInfoMap.getOrElse(userId, "")

        s"$userId,$userName,$liveName,${kv._2.size}"
      })
      .foreach(println)

  }

  @Test
  // 2、统计每位用户累计观看直播时长，按时长降序排列（6分）
  //  输出：[用户id，用户名，累计时长]
  def question02(): Unit = {
    val format: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    liveEvents
      .map(e => {
        val start: Date = format.parse(e.startTime)
        val end: Date = format.parse(e.endTime)
        val duration: Double = (end.getTime - start.getTime) / 1000 / 60.toDouble
        (e.userId, duration)
      })
      .groupBy(_._1)
      .map(kv => {
        val userName: String = userInfoMap.getOrElse(kv._1, "")
        val sumDuration: Double = kv._2.map(_._2).sum
        (kv._1, userName, sumDuration)
      })
      .toList
      .sortBy(-_._3)
      .foreach(println)

  }

  @Test
  // 3、统计不同类型直播用户累计观看时长降序排名（8分）
  // 输出：[直播间id，直播间类型，用户id，用户名，累计时长，排名]
  def question03(): Unit = {
    val format: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    liveEvents
      .map(e => {
        val start: Date = format.parse(e.startTime)
        val end: Date = format.parse(e.endTime)
        val duration: Double = (end.getTime - start.getTime) / 1000 / 60.toDouble
        (e.liveId, e.userId, duration)
      })
      // 合并每个用户在同一直播间观看的多次记录
      .groupBy(t3 => (t3._1, t3._2))
      .map(kv => {
        val liveId: String = kv._1._1
        val userId: String = kv._1._2
        (liveId, userId, kv._2.map(_._3).sum)
      })
      .groupBy(_._1)
      .flatMap(kv => {
        var rn: Int = 0
        kv._2
          .toList
          .sortBy(-_._3)
          .map(t3 => {
            rn += 1
            s"${t3._1},${liveTypesMap.getOrElse(t3._1,"")},${t3._2},${userInfoMap.getOrElse(t3._2,"")},${t3._3},$rn"
          })
      }).foreach(println)


  }
}
